#include "IMultiplex.h"
#include <assert.h>
#include <sys/epoll.h>
#include "Channel.h"
#include "Log.h"

namespace dsth {

class DSTHEpoll : public IMultiplex {
public:
    typedef std::vector<struct epoll_event> EventList;
    DSTHEpoll();
    virtual ~DSTHEpoll();
    virtual void UpdateChannel(Channel* channel);
    virtual void RemoveChannel(Channel* channel);
    virtual Timestamp Run(int time_out, IMultiplex::ChannelList* active_list);

private:
    // Can't invoke copy constructor or assignment operator
    DSTHEpoll(const DSTHEpoll&);
    void operator=(const DSTHEpoll&);

    static const int kInitEventListSize = 16;
    void UpdateEvent(int operation, Channel* channel);
    void UpdateEvent(int operation, Channel* channel,  int events);
    void DispatchEvent(int event_num, ChannelList* active_list);
    
    int epoll_;
    EventList events_;
};

IMultiplex* CreateMultiplexor() {
    return new DSTHEpoll;
}

DSTHEpoll::DSTHEpoll()
    : epoll_(::epoll_create1(EPOLL_CLOEXEC)),
    // for epoll_wait() buffer len
    events_(kInitEventListSize) {
    if (epoll_ < 0) {
        LOG_FATAL("call epoll_create1 failed");
    }
}

DSTHEpoll::~DSTHEpoll() {
    ::close(epoll_);
}

void DSTHEpoll::UpdateEvent(int operation, Channel* channel) {
    int events = 0;
    if (channel->IsReading()) {
        events |= (EPOLLIN | EPOLLPRI);
    }
    if (channel->IsWriting()) {
        events |= EPOLLOUT;
    }
    UpdateEvent(operation, channel, events);
}

void DSTHEpoll::UpdateEvent(int operation, Channel* channel,  int events) {
    struct epoll_event event;
    memset(&event, 0x00, sizeof event);
    event.events = events;
    event.data.ptr = channel;
    if (::epoll_ctl(epoll_, operation, channel->GetSocketFd(), &event) < 0) {
        LOG_FATAL("epoll_ctl failed, op=%d", operation);
    }
}
    
Timestamp DSTHEpoll::Run(int time_out, ChannelList* active_list) {
    int event_num = ::epoll_wait(epoll_, &*events_.begin(),
        static_cast<int>(events_.size()), time_out);
    Timestamp now_time(Timestamp::GetNowTime());
    if (event_num > 0) {
        DispatchEvent(event_num, active_list);
        if (event_num == static_cast<int>(events_.size())) {
            events_.resize(events_.size() * 2);
        }
    } else if (event_num == 0) {
        LOG_TRACE("[DSTHEpoll::Run] epoll_wait return 0");
    } else {
        LOG_ERROR("[DSTHEpoll::Run] epoll_wait return %d, errorcode:%d", event_num, errno);
    }
    return now_time;
}

void DSTHEpoll::DispatchEvent(int event_num, ChannelList* active_list) {
    assert(event_num <= static_cast<int>(events_.size()));
    for (int i = 0; i < event_num; ++i) {
        int event = events_[i].events;
        Channel* channel = static_cast<Channel*>(events_[i].data.ptr);
        assert(channel_map_.find(channel->GetSocketFd()) != channel_map_.end());
        int revent = 0;
        if ((event & EPOLLIN) || (event & EPOLLPRI)) {
            revent |= Channel::kReadEvent;
        }
        if (event & EPOLLOUT) {
            revent |= Channel::kWriteEvent;
        }
        if ((event & EPOLLRDHUP) || (event & EPOLLERR) || (event & EPOLLHUP)) {
            revent = Channel::kErrorEvent;
            LOG_ERROR("[DSTHEpoll::DispatchEvent] error event:%d", event);
        }
        assert(revent != 0);
        channel->SetRevent(revent);
        active_list->push_back(channel);
    }
}

void DSTHEpoll::UpdateChannel(Channel* channel) {
    assert(channel != NULL);
    int fd = channel->GetSocketFd();
    int event = channel->GetEvent();
    Channel::STATUS status = channel->GetStatus();
    if (status == Channel::E_NEW) {
        assert(event != Channel::kNoneEvent);
        assert(channel_map_.find(fd) == channel_map_.end());
        channel_map_[fd] = channel;
        UpdateEvent(EPOLL_CTL_ADD, channel);
        channel->SetStatus(Channel::E_ADDED);
        LOG_DEBUG("[DSTHEpoll::UpdateChannel] Add a new channel, fd:%d events:%d.", fd, event);
    }
    else if (status == Channel::E_DELED) {
        assert(event != Channel::kNoneEvent);
        assert(channel_map_.find(fd) != channel_map_.end());
        UpdateEvent(EPOLL_CTL_ADD, channel);
        channel->SetStatus(Channel::E_ADDED);
        LOG_DEBUG("[DSTHEpoll::UpdateChannel] Add a deleted channel, fd:%d events:%d.", fd, event);
    }
    else { // ADDED
        assert(channel_map_.find(fd) != channel_map_.end());
        if (event == Channel::kNoneEvent) {
            UpdateEvent(EPOLL_CTL_DEL, channel);
            channel->SetStatus(Channel::E_DELED);
            LOG_DEBUG("[DSTHEpoll::UpdateChannel] Deleted an added channel, fd:%d events:%d.", fd, event);
        }
        else {
            UpdateEvent(EPOLL_CTL_MOD, channel);
            LOG_DEBUG("[DSTHEpoll::UpdateChannel] Update an added channel, fd:%d events:%d.", fd, event);
        }
    }
}

void DSTHEpoll::RemoveChannel(Channel* channel) {
    assert(channel != NULL);
    int fd = channel->GetSocketFd();
    Channel::STATUS status = channel->GetStatus();
    assert(status == Channel::E_DELED || status == Channel::E_ADDED);
    assert(channel_map_.find(fd) != channel_map_.end());
    channel_map_.erase(fd);
    if (status == Channel::E_ADDED) {
        UpdateEvent(EPOLL_CTL_DEL, channel);
    }
    LOG_DEBUG("[DSTHEpoll::RemoveChannel] socket fd:%d.", fd);
}

} // namespace
